package xyz.scootaloo.kami.server.service.impl

import io.vertx.core.Future
import xyz.scootaloo.kami.server.service.Async
import xyz.scootaloo.kami.server.service.GeneralSyncService
import xyz.scootaloo.kami.server.service.GeneralSyncService.TopicTask
import xyz.scootaloo.kami.server.service.Sync
import java.util.*

/**
 * @author flutterdash@qq.com
 * @since 2022/3/25 13:33
 */
object InternalGeneralSyncServiceImpl : GeneralSyncService {

    override fun submit(t: TopicTask): Future<Unit> {
        return Controller.submit(t)
    }

    object Controller {
        private val topicMap = HashMap<String, TopicTaskQue>()

        fun submit(task: TopicTask): Future<Unit> = Sync.run {
            val que = getOrCreateTaskQue(task.topic)
            que.add(task)
            tryAcquire(que)
        }

        private fun release(topic: String) = Sync.run {
            val que = getOrCreateTaskQue(topic)
            que.available = true
            tryAcquire(que)
        }

        private fun tryAcquire(que: TopicTaskQue) {
            if (que.available && que.isNotEmpty()) {
                que.available = false
                val nextTask = que.remove()
                asyncRunCommand(nextTask)
            }
        }

        private fun asyncRunCommand(task: TopicTask) = Async.run {
            try {
                task.run()
            } finally {
                release(task.topic)
            }
        }

        private fun getOrCreateTaskQue(topic: String): TopicTaskQue {
            return topicMap[topic] ?: TopicTaskQue().apply {
                topicMap[topic] = this
            }
        }
    }

    class TopicTaskQue(
        var available: Boolean = true,
        private val queue: Queue<TopicTask> = LinkedList()
    ) : Queue<TopicTask> by queue
}